package com.max.zookeeper;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.*;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.Stat;

import java.util.concurrent.TimeUnit;

public class MyZookeeper {
    public static void main(String[] args) throws Exception {
        CuratorFramework curatorFramework= CuratorFrameworkFactory.
                //集群情况下的多个ZooKeeper之间使用逗号分隔
                        builder().connectString("master:2181").
                        sessionTimeoutMs(5000).retryPolicy(new
                        //链接重试
                        ExponentialBackoffRetry(1000,3)).
                        namespace("").build();
        //建立链接
        curatorFramework.start();
/*        InterProcessMutex lock = new InterProcessMutex(curatorFramework, "/lock");
        getLock(lock);
        System.out.println("doing......");
        unLock(lock);*/
//        add(curatorFramework);
        query(curatorFramework);
        update(curatorFramework);
        query(curatorFramework);
        curatorFramework.close();
    }

    public static void query(CuratorFramework curatorFramework) throws Exception {

        /*
        //监听
        Watcher w = new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                System.out.println("监听到的变化 watchedEvent = " + watchedEvent);
            }
        };
        byte[] content = curatorFramework.getData()
                .usingWatcher(w).forPath("/my");*/

        Stat stat=new Stat();
        //查询节点数据
        byte[] bytes = curatorFramework.getData().storingStatIn(stat).forPath("/hbase");
        System.out.println(new String(bytes));
        System.out.println(stat);
    }
    public static void delete(CuratorFramework curatorFramework) throws Exception {
        curatorFramework.delete().forPath("/my");
        // .deletingChildrenIfNeeded()//如果它有儿子都给删了
        //.withVersion() 根据版本号删
        System.out.println("delete ok");
    }

    public static void update(CuratorFramework curatorFramework) throws Exception {
        //curatorFramework.setData().withVersion(version).forPath(path);
        Stat stat= curatorFramework.setData().forPath("/my","new".getBytes());
        System.out.println(stat);
    }


    /*        另外还有
        PathChildrenCache子节点缓存用于子节点的监听，监控本节点的子节点被创建、
          更新或者删除.只能监听子节点，监听不到当前节点,不能递归监听，子节点下的子节点不能递归监控
        TreeCache不光能监听子节点，也能监听节点自身*/
    public static void watchNodeCache(CuratorFramework curatorFramework) throws Exception {
        //对单个节点创建watch事件
        //定义NodeCache,指明被监听节点的路径:用于监控本节点的新增，删除，更新
        NodeCache nodeCache = new NodeCache(curatorFramework,"/my");

        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                ChildData childData = nodeCache.getCurrentData();
                System.out.println(childData.getPath());
                System.out.println(new String(childData.getData()));
                System.out.println(childData.getStat());
            }});
        nodeCache.start(true);//开启true代表缓存当前节点
        Thread.sleep(60000);

    }

    public static void add(CuratorFramework curatorFramework) throws Exception {
        //状态
        Stat stat=new Stat();
        //是否存在
        System.out.println(curatorFramework.checkExists().forPath("/my"));
        //创建节点
        String s=curatorFramework
                .create()//创建节点
                //.creatingParentsIfNeeded()//创建父节点,如果需要的话
                .storingStatIn(stat)
                //.withMode(CreateMode.EPHEMERAL)//临时节点
                .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)//节点访问权限
                .forPath("/my","1235".getBytes());//指明你的节点路径,数据

        System.out.println(s);
        System.out.println(stat);
    }

    public static boolean getLock(InterProcessMutex lock) throws Exception {
        String lockNode = "/lock";
        /*InterProcessMutex：分布式可重入排它锁
        InterProcessSemaphoreMutex：分布式排它锁
        InterProcessReadWriteLock：分布式读写锁
        InterProcessMultiLock：将多个锁作为单个实体管理的容器*//*
         */
        //InterProcessMutex lock = new InterProcessMutex(curatorFramework, lockNode);
        return  lock.acquire(6000, TimeUnit.SECONDS) ;

    }

    public static void unLock(InterProcessMutex lock) throws Exception {
        lock.release();
    }
}
